//package com.example.demo.demos.service.impl;
//
//import com.fasterxml.jackson.core.JsonFactory;
//import com.fasterxml.jackson.core.JsonParser;
//import com.fasterxml.jackson.core.JsonToken;
//import com.fasterxml.jackson.databind.DeserializationFeature;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.jcraft.jsch.*;
//import com.zaxxer.hikari.HikariConfig;
//import com.zaxxer.hikari.HikariDataSource;
//import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
//import org.apache.commons.pool2.BasePooledObjectFactory;
//import org.apache.commons.pool2.PooledObject;
//import org.apache.commons.pool2.impl.DefaultPooledObject;
//import org.apache.commons.pool2.impl.GenericObjectPool;
//import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.stereotype.Service;
//
//import javax.annotation.PostConstruct;
//import javax.annotation.PreDestroy;
//import java.io.*;
//import java.math.BigDecimal;
//import java.nio.charset.StandardCharsets;
//import java.nio.file.Files;
//import java.nio.file.Path;
//import java.nio.file.Paths;
//import java.sql.*;
//import java.sql.Date;
//import java.util.*;
//import java.util.concurrent.atomic.AtomicLong;
//import java.util.stream.Collectors;
//
//@Service
//public class SftpSyncService {
//    private static final Logger logger = LoggerFactory.getLogger(SftpSyncService.class);
//    private static final String DEFAULT_PK_FIELD = "id";
//    private static final int MAX_RETRIES = 3;
//    private static final long RETRY_DELAY_MS = 1000;
//    private static final int BATCH_RETRY_COUNT = 2;
//
//    // 连接池监控计数
//    private final AtomicLong borrowedCount = new AtomicLong(0);
//    private final AtomicLong returnedCount = new AtomicLong(0);
//    private final AtomicLong createdCount = new AtomicLong(0);
//    private final AtomicLong destroyedCount = new AtomicLong(0);
//
//    // 配置参数
//    @Value("${sftp.host}")
//    private String host;
//    @Value("${sftp.port}")
//    private int port;
//    @Value("${sftp.username}")
//    private String username;
//    @Value("${sftp.password}")
//    private String password;
//    @Value("${sftp.privateKey}")
//    private String privateKey;
//    @Value("${sftp.passphrase}")
//    private String passphrase;
//    @Value("${sftp.remotePath}")
//    private String remotePath;
//    @Value("${sftp.localTempPath}")
//    private String localTempPath;
//    @Value("${sftp.sessionTimeout}")
//    private int sessionTimeout;
//    @Value("${sftp.channelTimeout}")
//    private int channelTimeout;
//
//    @Value("${db.url}")
//    private String dbUrl;
//    @Value("${db.username}")
//    private String dbUsername;
//    @Value("${db.password}")
//    private String dbPassword;
//    @Value("${batch.size:500}")
//    private int batchSize;
//    @Value("${isSynchronousAll}")
//    private int isSynchronousAll;
//
//    // Jackson对象（线程安全）
//    private final ObjectMapper objectMapper = new ObjectMapper()
//            .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
//
//    // 连接池和状态管理
//    private GenericObjectPool<ChannelSftp> sftpPool;
//    private Session session;
//    private JSch jsch;
//    private final Map<String, Map<String, Integer>> tableSchemaCache = new HashMap<>();
//    private static Integer countNumber = 0;
//    private volatile boolean isRunning = false;
//
//    private static final String[] EXCLUDED_TABLES = {
//
//            "annual_report_asset",
//            "annual_report_invest",
//            "bankruptcy_public_announcement",
//            "company_base",
//            "company_business_import_export_credit",
//            "company_chattel_mortgage",
//            "company_employee",
//            "company_industry",
//            "company_justice",
//            "company_justice_unfreeze",
//            "company_partner_pay",
//            "company_punishment_info_creditchina_new",
//            "company_simple_cancel_objection",
//            "ip_copyright_works",
//            "ip_trademark_applicant_list",
//            "opt_ent_partner_cn",
//            "qimingpian_product",
//            "risk_court_filed_case_litigant",
//            "risk_environment_punish",
//            "risk_restricted_outbound",
//            "risk_zhongben",
//            "special_hongkong_company",
//            "stock_allotment",
//            "stock_equity_structure",
//            "stock_issue",
//            "tb_company_sme",
//            "tb_project_jingpin",
//
//
//
//            "annual_report_base",
//            "annual_report_partner",
//            "bankruptcy_public_cases",
//            "company_base_clean",
//            "company_cancel_notice",
//            "company_chattel_pawn",
//            "company_employment",
//            "company_industry_model",
//            "company_justice_equity_change",
//            "company_license_info_creditchina_new",
//            "company_partner_realpay",
//            "company_qy_partner",
//            "company_stock_change",
//            "ip_copyright_works_list",
//            "ip_trademark_category_list",
//            "opt_ip_patent_affair_status",
//            "qimingpian_team_member",
//            "risk_court_notice",
//            "risk_evaluate_org",
//            "risk_shixin",
//            "sf_cpws",
//            "special_hongkong_company_history",
//            "stock_announcement",
//            "stock_executive",
//            "sys_cat",
//            "tb_general_taxpayer",
//            "tm_class_dict",
//
//
//
//
//            "annual_report_change",
//            "annual_report_social_security",
//            "bankruptcy_public_cases_list",
//            "company_bid_entity_list_new",
//            "company_change",
//            "company_check",
//            "company_history_name",
//            "company_intellectual",
//            "company_justice_freeze",
//            "company_logo",
//            "company_pledge",
//            "company_random_check",
//            "investment_event",
//            "ip_patent",
//            "ip_trademark_flow_list",
//            "opt_ip_patent_info",
//            "risk_court_announcement",
//            "risk_court_notice_list",
//            "risk_evaluate_org_list",
//            "risk_tax_punish",
//            "sf_cpws_dsr",
//            "special_law_office",
//            "stock_base",
//            "stock_financial_analysis",
//            "tax_credit",
//            "tb_icp_base_info",
//            "tradable_stock_holder",
//
//
//
//            "annual_report_equity_change",
//            "annual_report_website",
//            "company_abnormal",
//            "company_bid_main",
//            "company_chattel",
//            "company_clear",
//            "company_illegal",
//            "company_intellectual_change",
//            "company_justice_invalid",
//            "company_own_tax_info",
//            "company_profile",
//            "company_random_check_result",
//            "ip_copyright_software",
//            "ip_patent_applicant_list",
//            "land_publicity",
//            "opt_ip_patent_patentee",
//            "risk_court_announcement_list",
//            "risk_court_service_announcement",
//            "risk_evaluate_result",
//            "risk_xianxiao",
//            "special_enterprise",
//            "special_social_organ",
//            "stock_bonus",
//            "stock_holder",
//            "tb_certificate_detail_integrate",
//            "tb_judicial_sale",
//
//
//
//            "annual_report_guarantee",
//            "bankruptcy_judicative_paper",
//            "company_allow",
//            "company_branch",
//            "company_chattel_change",
//            "company_customs_import_export_credit_rating",
//            "company_import_export_administr_penalty",
//            "company_investor",
//            "company_justice_keep_freeze",
//            "company_partner",
//            "company_punish",
//            "company_simple_cancel",
//            "ip_copyright_software_list",
//            "ip_trademark",
//            "land_result_announcement",
//            "qimingpian_history_rongzi",
//            "risk_court_filed_case",
//            "risk_court_service_announcement_litigant",
//            "risk_evaluate_result_list",
//            "risk_zhixing",
//            "special_gov_unit",
//            "special_trade_union",
//            "stock_equity_change",
//            "stock_holding",
//            "tb_certificate_integrate",
//            "tb_judicial_sale_info_company",
//    };
//
//    // 配置连接池
//    private static HikariDataSource dataSource;
//
//    @PostConstruct
//    public void initDataSource() {
//        HikariConfig config = new HikariConfig();
//        config.setJdbcUrl(dbUrl);
//        config.setUsername(dbUsername);
//        config.setPassword(dbPassword);
//        config.setMaximumPoolSize(15); // 根据需求调整
//        config.setConnectionTimeout(30000); // 30秒
//        config.setIdleTimeout(600000); // 10分钟空闲超时
//        config.setMaxLifetime(1800000); // 30分钟最大生命周期
//        config.setAutoCommit(false); // 与您现有的事务管理一致
//
//        // 连接测试查询
//        config.setConnectionTestQuery("SELECT 1");
//
//        dataSource = new HikariDataSource(config);
//    }
//
//    // 获取连接的方法
//    private Connection getConnection() throws SQLException {
//        Connection conn = dataSource.getConnection();
//        conn.setAutoCommit(false); // 保持您现有的事务行为
//        return conn;
//    }
//
//    // 连接池工厂
//    private class SftpChannelFactory extends BasePooledObjectFactory<ChannelSftp> {
//        @Override
//        public ChannelSftp create() throws Exception {
//            ChannelSftp channel = (ChannelSftp) session.openChannel("sftp");
//            channel.connect(channelTimeout);
//            createdCount.incrementAndGet();
//            logger.debug("Created new SFTP channel: {}", channel);
//            return channel;
//        }
//
//        @Override
//        public PooledObject<ChannelSftp> wrap(ChannelSftp channel) {
//            return new DefaultPooledObject<>(channel);
//        }
//
//        @Override
//        public void destroyObject(PooledObject<ChannelSftp> p) throws Exception {
//            if (p.getObject().isConnected()) {
//                p.getObject().disconnect();
//                destroyedCount.incrementAndGet();
//                logger.debug("Destroyed SFTP channel: {}", p.getObject());
//            }
//        }
//
//        @Override
//        public boolean validateObject(PooledObject<ChannelSftp> p) {
//            boolean isValid = p.getObject().isConnected();
//            if (!isValid) {
//                logger.warn("Invalid SFTP channel detected: {}", p.getObject());
//            }
//            return isValid;
//        }
//    }
//
//    @PostConstruct
//    public void init() throws Exception {
//        validateAndCreateDirectory(localTempPath);
//        initSftpSession();
//        initConnectionPool();
//        logger.info("SFTP同步服务初始化完成. 连接池配置: maxTotal={}, minIdle={}, maxIdle={}",
//                sftpPool.getMaxTotal(), sftpPool.getMinIdle(), sftpPool.getMaxIdle());
//    }
//
//    private void initSftpSession() throws JSchException {
//        jsch = new JSch();
//
//        if (privateKey != null && !privateKey.isEmpty()) {
//            if (passphrase != null && !passphrase.isEmpty()) {
//                jsch.addIdentity(privateKey, passphrase);
//            } else {
//                jsch.addIdentity(privateKey);
//            }
//        }
//
//        session = jsch.getSession(username, host, port);
//        if (password != null && !password.isEmpty()) {
//            session.setPassword(password);
//        }
//
//        Properties config = new Properties();
//        config.put("StrictHostKeyChecking", "no");
//        session.setConfig(config);
//        session.setTimeout(sessionTimeout);
//        session.connect();
//        logger.info("已建立SFTP会话: {}@{}:{}", username, host, port);
//    }
//
//    private void initConnectionPool() {
//        GenericObjectPoolConfig<ChannelSftp> config = new GenericObjectPoolConfig<>();
//        config.setMaxTotal(5);  // 减少最大连接数避免资源争用
//        config.setMinIdle(2);
//        config.setMaxIdle(3);   // 降低最大空闲连接数
//        config.setTestOnBorrow(true);
//        config.setTestWhileIdle(true);
//        config.setMaxWaitMillis(5000);
//        config.setTimeBetweenEvictionRunsMillis(30000);
//
//        sftpPool = new GenericObjectPool<>(new SftpChannelFactory(), config);
//        logger.info("SFTP连接池初始化完成");
//    }
//
//    private void validateAndCreateDirectory(String path) throws IOException {
//        Path dirPath = Paths.get(path);
//        if (!Files.exists(dirPath)) {
//            Files.createDirectories(dirPath);
//            logger.info("已创建目录: {}", dirPath);
//        }
//        if (!Files.isWritable(dirPath)) {
//            throw new IOException("目录不可写: " + dirPath);
//        }
//    }
//
//    @PreDestroy
//    public void cleanup() {
//        logConnectionPoolStats();
//
//        if (sftpPool != null && !sftpPool.isClosed()) {
//            sftpPool.close();
//            logger.info("SFTP连接池已关闭");
//        }
//        if (session != null && session.isConnected()) {
//            session.disconnect();
//            logger.info("SFTP会话已断开");
//        }
//    }
//
//    private void logConnectionPoolStats() {
//        if (sftpPool != null) {
//            logger.info("连接池统计: 创建={}, 销毁={}, 借用={}, 归还={}, 活跃={}, 空闲={}",
//                    createdCount.get(),
//                    destroyedCount.get(),
//                    borrowedCount.get(),
//                    returnedCount.get(),
//                    sftpPool.getNumActive(),
//                    sftpPool.getNumIdle());
//        }
//    }
//
//    public void syncAllTables() {
//        if (isRunning) {
//            logger.warn("上一次同步任务仍在执行，跳过本次执行");
//            return;
//        }
//        isRunning = true;
//
//        try (Connection conn = getConnection()){
//            // 实际数据处理（每个文件独立事务）
//            processRemoteDirectories(conn);
//        } catch (Exception e) {
//            logger.error("同步任务失败", e);
//        } finally {
//            isRunning = false;
//        }
//    }
//
//    private void processRemoteDirectories(Connection conn) throws Exception {
//        List<String> tableDirs = listRemoteDirectories(remotePath);
//        for (String tableDir : tableDirs) {
//            try {
//                processTableDirectory(conn, tableDir);
//            } catch (Exception e) {
//                logger.error("处理表目录 {} 失败，跳过继续处理其他表: {}", tableDir, e.getMessage(), e);
//            }
//        }
//        countNumber += 2;
//    }
//
//    private List<String> listRemoteDirectories(String path) throws Exception {
//        ChannelSftp channel = borrowChannel();
//        try {
//            return safeListFiles(channel, path).stream()
//                    .filter(entry -> !isSpecialDirectory(entry.getFilename()))
//                    .filter(entry -> entry.getAttrs().isDir())
//                    .map(ChannelSftp.LsEntry::getFilename)
//                    .collect(Collectors.toList());
//        } finally {
//            returnChannel(channel);
//        }
//    }
//
//    private List<ChannelSftp.LsEntry> safeListFiles(ChannelSftp channel, String path) throws SftpException {
//        Vector<?> fileList = channel.ls(path);
//        List<ChannelSftp.LsEntry> entries = new ArrayList<>();
//
//        for (Object item : fileList) {
//            if (item instanceof ChannelSftp.LsEntry) {
//                entries.add((ChannelSftp.LsEntry) item);
//            } else {
//                logger.warn("意外的SFTP列表项类型: {}", item.getClass());
//            }
//        }
//        return entries;
//    }
//
//    private boolean isSpecialDirectory(String filename) {
//        return filename.equals(".") || filename.equals("..");
//    }
//
//    private void processTableDirectory(Connection conn, String tableDir) throws Exception {
//        if (!Arrays.asList(EXCLUDED_TABLES).contains(tableDir)) {
//            logger.info("跳过 {} 表的同步", tableDir);
//            return;
//        }
//        String tableName = shouldAppendYbSuffix(tableDir) ? tableDir + "_yb" : tableDir;
//        logger.info("正在处理表目录: {}", tableDir);
//
//        loadTableSchema(conn, tableName);
//
//        if ("company_employee".equals(tableDir) && isSynchronousAll == countNumber) {
//            logger.info("特殊处理: company_employee表同步20250211及以后的数据");
//            List<String> dateDirs = listRemoteDirectories(remotePath + tableDir + "/").stream()
//                    .filter(dir -> dir.compareTo("20250211") >= 0)
//                    .collect(Collectors.toList());
//
//            for (String dateDir : dateDirs) {
//                processDateDirectory(tableName, dateDir, tableDir);
//            }
//        } else if ("company_base_clean".equals(tableDir) && isSynchronousAll == countNumber) {
//            logger.info("特殊处理: company_base_clean表同步20241210及以后的数据");
//            List<String> dateDirs = listRemoteDirectories(remotePath + tableDir + "/").stream()
//                    .filter(dir -> dir.compareTo("20241210") >= 0)
//                    .collect(Collectors.toList());
//
//            for (String dateDir : dateDirs) {
//                processDateDirectory(tableName, dateDir, tableDir);
//            }
//        } else if ("opt_ip_patent_info".equals(tableDir) && isSynchronousAll == countNumber) {
//            logger.info("特殊处理: opt_ip_patent_info表同步20250125及以后的数据");
//            List<String> dateDirs = listRemoteDirectories(remotePath + tableDir + "/").stream()
//                    .filter(dir -> dir.compareTo("20250125") >= 0)
//                    .collect(Collectors.toList());
//
//            for (String dateDir : dateDirs) {
//                processDateDirectory(tableName, dateDir, tableDir);
//            }
//        } else if ("company_abnormal".equals(tableDir) && isSynchronousAll == countNumber) {
//            logger.info("特殊处理: company_abnormal表同步20241224及以后的数据");
//            List<String> dateDirs = listRemoteDirectories(remotePath + tableDir + "/").stream()
//                    .filter(dir -> dir.compareTo("20241224") >= 0)
//                    .collect(Collectors.toList());
//
//            for (String dateDir : dateDirs) {
//                processDateDirectory(tableName, dateDir, tableDir);
//            }
//        } else if ("risk_zhixing".equals(tableDir) && isSynchronousAll == countNumber) {
//            logger.info("特殊处理: risk_zhixing表同步20250101及以后的数据");
//            List<String> dateDirs = listRemoteDirectories(remotePath + tableDir + "/").stream()
//                    .filter(dir -> dir.compareTo("20250101") >= 0)
//                    .collect(Collectors.toList());
//
//            for (String dateDir : dateDirs) {
//                processDateDirectory(tableName, dateDir, tableDir);
//            }
//        } else if ("company_base".equals(tableDir) && isSynchronousAll == countNumber) {
//            logger.info("特殊处理: company_base表同步20241103及以后的数据");
//            List<String> dateDirs = listRemoteDirectories(remotePath + tableDir + "/").stream()
//                    .filter(dir -> dir.compareTo("20241103") >= 0)
//                    .collect(Collectors.toList());
//
//            for (String dateDir : dateDirs) {
//                processDateDirectory(tableName, dateDir, tableDir);
//            }
//        } else if (isSynchronousAll == countNumber) {
//            logger.info("同步所有数据");
//            List<String> dateDirs = listRemoteDirectories(remotePath + tableDir + "/");
//            for (String dateDir : dateDirs) {
//                processDateDirectory(tableName, dateDir, tableDir);
//            }
//        } else {
//            logger.info("同步最新日期数据");
//            List<String> dateDirs = listRemoteDirectories(remotePath + tableDir + "/")
//                    .stream()
//                    .sorted(Comparator.reverseOrder())
//                    .collect(Collectors.toList());
//
//            if (!dateDirs.isEmpty()) {
//                processDateDirectory(tableName, dateDirs.get(0), tableDir);
//            }
//        }
//    }
//
//    private boolean shouldAppendYbSuffix(String tableName) {
//        return new HashSet<>(Arrays.asList(
//                "company_change",
//                "company_industry",
//                "company_history_name",
//                "company_partner_realpay",
//                "company_partner",
//                "company_partner_pay",
//                "tb_company_sme"
//        )).contains(tableName);
//    }
//
//    private void loadTableSchema(Connection conn, String tableName) throws SQLException {
//        if (!tableSchemaCache.containsKey(tableName)) {
//            Map<String, Integer> schema = new HashMap<>();
//            try (ResultSet rs = conn.getMetaData().getColumns(null, null, tableName, null)) {
//                while (rs.next()) {
//                    schema.put(rs.getString("COLUMN_NAME"), rs.getInt("DATA_TYPE"));
//                }
//            }
//
//            if (!schema.containsKey(DEFAULT_PK_FIELD)) {
//                throw new SQLException("缺少主键字段: " + DEFAULT_PK_FIELD);
//            }
//
//            tableSchemaCache.put(tableName, schema);
//            logger.info("已加载表结构: {} ({}列)", tableName, schema.size());
//        }
//    }
//
//    private void processDateDirectory(String tableName, String dateDir, String tableDir) throws Exception {
//        logger.debug("开始处理日期目录: {}/{}", tableDir, dateDir);
//        ChannelSftp channel = borrowChannel();
//        try {
//            List<String> gzFiles = safeListFiles(channel, remotePath + tableDir + "/" + dateDir + "/").stream()
//                    .filter(entry -> !isSpecialDirectory(entry.getFilename()))
//                    .filter(entry -> !entry.getAttrs().isDir())
//                    .filter(entry -> entry.getFilename().endsWith(".gz"))
//                    .map(ChannelSftp.LsEntry::getFilename)
//                    .collect(Collectors.toList());
//
//            for (String gzFile : gzFiles) {
//                try {
//                    // 每个文件独立处理
//                    processGzFile(tableName, remotePath + tableDir + "/" + dateDir + "/" + gzFile);
//                } catch (Exception e) {
//                    logger.error("处理文件 {} 失败，跳过继续处理其他文件: {}", gzFile, e.getMessage(), e);
//                }
//            }
//        } finally {
//            returnChannel(channel);
//        }
//        logger.debug("完成处理日期目录: {}/{}", tableDir, dateDir);
//    }
//
//    private void processGzFile(String tableName, String remoteFilePath) throws Exception {
//        long startTime = System.currentTimeMillis();
//        File localFile = downloadFileWithRetry(remoteFilePath);
//
//        // 每个文件使用独立的连接和事务
//        try (Connection conn = getConnection()) {
//            conn.setAutoCommit(false);
//            try {
//                logger.info("开始处理文件: {}, 大小: {}MB", remoteFilePath,
//                        Files.size(localFile.toPath()) / (1024 * 1024));
//
//                processJsonStream(conn, tableName, localFile);
//                conn.commit();
//                long parseTime = System.currentTimeMillis() - startTime;
//                logger.info("文件处理完成: {} (总耗时: {}ms)", remoteFilePath, parseTime);
//            } catch (Exception e) {
//                conn.rollback();
//                logger.error("文件处理失败（已回滚）: {}", remoteFilePath, e);
//                throw e;
//            }
//        } finally {
//            Files.deleteIfExists(localFile.toPath());
//        }
//    }
//
//    // 流式解析JSON的核心方法（大幅降低内存使用）
//    private void processJsonStream(Connection conn, String tableName, File localFile) throws Exception {
//        Map<String, Integer> schema = tableSchemaCache.get(tableName);
//        if (schema == null) {
//            throw new SQLException("未找到表结构: " + tableName);
//        }
//
//        List<Map<String, Object>> batchRecords = new ArrayList<>(batchSize);
//        JsonFactory factory = objectMapper.getFactory();
//
//        try (InputStream fis = new FileInputStream(localFile);
//             InputStream gzis = new GzipCompressorInputStream(fis);
//             BufferedReader reader = new BufferedReader(new InputStreamReader(gzis, StandardCharsets.UTF_8))) { // 使用BufferedReader逐行读取
//
//            int recordCount = 0;
//            String line;
//            while ((line = reader.readLine()) != null) {
//                try {
//                    // 使用Jackson解析单行JSON对象
//                    try (JsonParser parser = factory.createParser(line)) {
//                        JsonToken token = parser.nextToken();
//                        if (token != JsonToken.START_OBJECT) {
//                            logger.warn("跳过非JSON对象行: {}", line);
//                            continue;
//                        }
//
//                        Map<String, Object> record = parseJsonObject(parser, schema);
//                        if (record.containsKey(DEFAULT_PK_FIELD)) {
//                            recordCount++;
//                            Map<String, Object> sanitizedRecord = sanitizeRecord(record);
//                            batchRecords.add(sanitizedRecord);
//
//                            if (batchRecords.size() >= batchSize) {
//                                executeUpsertWithRetry(conn, tableName, batchRecords, BATCH_RETRY_COUNT);
//                                batchRecords.clear();
//                                logMemoryUsage("批处理后");
//                            }
//                        }
//                    }
//                } catch (Exception e) {
////                    logger.warn("解析行失败: {} | 错误: {}", line, e.getMessage());
//                }
//            }
//
//            // 处理剩余记录
//            if (!batchRecords.isEmpty()) {
//                executeUpsertWithRetry(conn, tableName, batchRecords, BATCH_RETRY_COUNT);
//                batchRecords.clear();
//            }
//
//            logger.info("解析完成: {}条记录", recordCount);
//        }
//    }
//
//    private Map<String, Object> parseJsonObject(JsonParser parser, Map<String, Integer> schema) throws IOException {
//        Map<String, Object> record = new HashMap<>();
//        while (parser.nextToken() != JsonToken.END_OBJECT) {
//            String field = parser.currentName();
//            parser.nextToken(); // 移动到值
//
//            if (schema.containsKey(field)) {
//                record.put(field, convertJsonValue(parser, schema.get(field)));
//            }
//        }
//        return record;
//    }
//
//    private Object convertJsonValue(JsonParser parser, int sqlType) throws IOException {
//        if (parser.currentToken() == JsonToken.VALUE_NULL) {
//            return null;
//        }
//
//        try {
//            switch (sqlType) {
//                case Types.SMALLINT:
//                case Types.INTEGER:
//                    return parser.getIntValue();
//                case Types.BIGINT:
//                    return parser.getLongValue();
//                case Types.FLOAT:
//                case Types.DOUBLE:
//                    return parser.getDoubleValue();
//                case Types.DECIMAL:
//                case Types.NUMERIC:
//                    return new BigDecimal(parser.getValueAsString());
//                case Types.BOOLEAN:
//                    return parser.getBooleanValue();
//                case Types.DATE:
//                    return Date.valueOf(parser.getValueAsString().split("T")[0]);
//                case Types.TIMESTAMP:
//                    return Timestamp.valueOf(parser.getValueAsString().replace("T", " "));
//                default:
//                    return parser.getValueAsString();
//            }
//        } catch (Exception e) {
//            logger.warn("值转换失败: {}", e.getMessage());
//            return null;
//        }
//    }
//
//    private File downloadFileWithRetry(String remotePath) throws Exception {
//        int retryCount = 0;
//        Exception lastException = null;
//
//        while (retryCount < MAX_RETRIES) {
//            ChannelSftp channel = borrowChannel();
//            try {
//                String fileName = Paths.get(remotePath).getFileName().toString();
//                File localFile = new File(localTempPath + fileName);
//
//                try (OutputStream out = new FileOutputStream(localFile)) {
//                    channel.get(remotePath, out);
//                    logger.debug("已下载文件: {} 到 {}", remotePath, localFile);
//                    return localFile;
//                }
//            } catch (SftpException e) {
//                lastException = e;
//                if (e.id == ChannelSftp.SSH_FX_NO_SUCH_FILE) {
//                    throw new FileNotFoundException("远程文件不存在: " + remotePath);
//                }
//                retryCount++;
//                logger.warn("文件下载失败，第{}/{}次重试: {}", retryCount, MAX_RETRIES, e.getMessage());
//                Thread.sleep(RETRY_DELAY_MS * retryCount);
//            } finally {
//                returnChannel(channel);
//            }
//        }
//        throw new RuntimeException("文件下载失败，已达最大重试次数", lastException);
//    }
//
//    private Map<String, Object> sanitizeRecord(Map<String, Object> record) {
//        Map<String, Object> sanitized = new HashMap<>(record.size());
//
//        record.forEach((key, value) -> {
//            if (value instanceof String) {
//                sanitized.put(key, sanitizeString((String) value));
//            } else {
//                sanitized.put(key, value);
//            }
//        });
//
//        return sanitized;
//    }
//
//    private String sanitizeString(String input) {
//        if (input == null) {
//            return null;
//        }
//
//        // 使用更高效的字符检查方式
//        int len = input.length();
//        int i = 0;
//        while (i < len) {
//            char c = input.charAt(i);
//            if (c <= 31 || c == 127) break;
//            i++;
//        }
//
//        // 不需要清理则直接返回修剪后的字符串
//        if (i == len) {
//            return input.trim();
//        }
//
//        // 需要清理时才进行实际处理
//        char[] chars = input.toCharArray();
//        for (; i < chars.length; i++) {
//            if (chars[i] <= 31 || chars[i] == 127) {
//                chars[i] = '?';
//            }
//        }
//        return new String(chars).trim();
//    }
//
//    private void executeUpsertWithRetry(Connection conn, String tableName, List<Map<String, Object>> records, int maxRetries) throws SQLException {
//        if (records.isEmpty()) return;
//
//        int retryCount = 0;
//        while (retryCount <= maxRetries) {
//            try {
//                executeUpsert(conn, tableName, records);
//                return;
//            } catch (SQLException e) {
//                retryCount++;
//                if (retryCount > maxRetries) {
//                    logger.error("表 {} 数据插入失败，已达最大重试次数: {}", tableName, e.getMessage(), e);
//                    throw e;
//                }
//                logger.warn("表 {} 数据插入失败，第{}/{}次重试: {}", tableName, retryCount, maxRetries, e.getMessage());
//                try {
//                    Thread.sleep(1000 * retryCount);
//                } catch (InterruptedException ie) {
//                    Thread.currentThread().interrupt();
//                    throw new SQLException("操作被中断", ie);
//                }
//            }
//        }
//    }
//
//    private void executeUpsert(Connection conn, String tableName, List<Map<String, Object>> records) throws SQLException {
//        if (records.isEmpty()) return;
//
//        String[] fields = records.get(0).keySet().toArray(new String[0]);
//        String sql = buildUpsertSql(conn, tableName, fields);
//
//        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
//            int totalProcessed = 0;
//            for (Map<String, Object> record : records) {
//                try {
//                    for (int i = 0; i < fields.length; i++) {
//                        setStatementParameter(stmt, i + 1, record.get(fields[i]),
//                                tableSchemaCache.get(tableName).get(fields[i]));
//                    }
//                    stmt.addBatch();
//                    totalProcessed++;
//
//                    // 分批执行，避免超大事务
//                    if (totalProcessed % batchSize == 0) {
//                        executeBatch(stmt, tableName, totalProcessed);
//                    }
//                } catch (SQLException e) {
//                    logger.warn("表 {} 记录处理失败，跳过此记录: {}", tableName, e.getMessage());
//                }
//            }
//
//            // 处理剩余批次
//            if (totalProcessed % batchSize != 0) {
//                executeBatch(stmt, tableName, totalProcessed);
//            }
//        }
//    }
//
//    private void executeBatch(PreparedStatement stmt, String tableName, int totalCount) throws SQLException {
//        try {
//            int[] results = stmt.executeBatch();
//            int successCount = 0;
//            for (int result : results) {
//                if (result >= 0 || result == Statement.SUCCESS_NO_INFO) {
//                    successCount++;
//                }
//            }
//            logger.info("表 {} 批处理结果: 成功={}/总数={}", tableName, successCount, totalCount);
//        } catch (BatchUpdateException e) {
//            handleBatchUpdateException(e, tableName);
//        } finally {
//            stmt.clearBatch();
//        }
//    }
//
//    private void handleBatchUpdateException(BatchUpdateException e, String tableName) throws SQLException {
//        int[] updateCounts = e.getUpdateCounts();
//        int failedCount = 0;
//
//        for (int i = 0; i < updateCounts.length; i++) {
//            if (updateCounts[i] == Statement.EXECUTE_FAILED) {
//                failedCount++;
//            }
//        }
//
//        logger.warn("表 {} 批量插入部分失败: 总条数={}, 失败数={}",
//                tableName, updateCounts.length, failedCount);
//
//        SQLException next = e.getNextException();
//        while (next != null) {
//            logger.warn("插入错误详情: {}", next.getMessage());
//            next = next.getNextException();
//        }
//    }
//
//    private String buildUpsertSql(Connection conn, String tableName, String[] fields) throws SQLException {
//        if (isPostgreSQL(conn)) {
//            return buildPostgresUpsertSql(tableName, fields);
//        } else {
//            return buildGenericUpsertSql(tableName, fields);
//        }
//    }
//
//    private String buildPostgresUpsertSql(String tableName, String[] fields) {
//        String columns = String.join(",", fields);
//        String placeholders = String.join(",", Collections.nCopies(fields.length, "?"));
//        String updates = Arrays.stream(fields)
//                .filter(f -> !f.equals(DEFAULT_PK_FIELD))
//                .map(f -> f + "=EXCLUDED." + f)
//                .collect(Collectors.joining(","));
//
//        return String.format(
//                "INSERT INTO %s (%s) VALUES (%s) ON CONFLICT (%s) DO UPDATE SET %s",
//                tableName, columns, placeholders, DEFAULT_PK_FIELD, updates);
//    }
//
//    private String buildGenericUpsertSql(String tableName, String[] fields) {
//        String columns = String.join(",", fields);
//        String placeholders = String.join(",", Collections.nCopies(fields.length, "?"));
//        String updates = Arrays.stream(fields)
//                .filter(f -> !f.equals(DEFAULT_PK_FIELD))
//                .map(f -> f + "=?")
//                .collect(Collectors.joining(","));
//
//        return String.format(
//                "MERGE INTO %s USING (VALUES(%s)) AS tmp(col) ON %s.%s=tmp.col " +
//                        "WHEN MATCHED THEN UPDATE SET %s WHEN NOT MATCHED THEN INSERT (%s) VALUES (%s)",
//                tableName, placeholders, tableName, DEFAULT_PK_FIELD, updates, columns, placeholders);
//    }
//
//    private void setStatementParameter(PreparedStatement stmt, int index, Object value, int sqlType) throws SQLException {
//        if (value == null) {
//            stmt.setNull(index, sqlType);
//        } else {
//            stmt.setObject(index, value, sqlType);
//        }
//    }
//
//    private boolean isPostgreSQL(Connection conn) throws SQLException {
//        return conn.getMetaData().getDatabaseProductName().equals("PostgreSQL");
//    }
//
//    private ChannelSftp borrowChannel() throws Exception {
//        borrowedCount.incrementAndGet();
//        return sftpPool.borrowObject();
//    }
//
//    private void returnChannel(ChannelSftp channel) {
//        if (channel != null) {
//            returnedCount.incrementAndGet();
//            sftpPool.returnObject(channel);
//        }
//    }
//
//    // 内存监控方法
//    private void logMemoryUsage(String context) {
//        if (logger.isDebugEnabled()) {
//            Runtime runtime = Runtime.getRuntime();
//            long usedMB = (runtime.totalMemory() - runtime.freeMemory()) / (1024 * 1024);
//            long totalMB = runtime.totalMemory() / (1024 * 1024);
//            long maxMB = runtime.maxMemory() / (1024 * 1024);
//            logger.debug("内存状态[{}]: 已用={}MB, 分配={}MB, 最大={}MB",
//                    context, usedMB, totalMB, maxMB);
//        }
//    }
//}
